package com.changan.utils;

import com.jayway.jsonpath.JsonPath;

import java.util.HashMap;
import java.util.List;
import java.util.Properties;

/**
 * @author roohom
 * @date 2021/8/18
 * <p>
 * flink restApi相关操作
 */
public class FlinkRestApiUtil {

    /**
     * 获取基础的baseUrl
     *
     * @param host     主机地址
     * @param port     端口号
     * @param elements 参数
     * @return 任务的 url
     */
    public static String baseUrl(String host, String port, String... elements) {
        return String.format("http://%s:%s/", host, port) + String.join("/", elements);
    }

    /**
     * 取消掉这个flink job
     *
     * @param host  主机地址
     * @param port  端口
     * @param jobId Flink的jobId
     * @return 返回flink 的job id
     */
    public static Integer terminateFlinkJobByJobId(String host, String port, String jobId) {
        return HttpUtils.httpPatch(baseUrl(host, port, "jobs", jobId));
    }

    /**
     * 取消掉这个flink job, 同时生成savepoint
     * 会返回一个triggerid用来查询状态, IN_PROGRESS -> COMPLETED <br/>
     * 任务已经被cancel, savepoint会一直制作,知道制作成功,并且期间一直可以查询savepoint状态 </p>
     * 返回示例: {"status":{"id":"COMPLETED"},"operation":{"location":"hdfs://nameservice1/tmp/yarnjobs/savepoint-05b1a1-cb29b9f330c7"}}
     * @param host  主机地址
     * @param port  端口
     * @param jobId Flink的jobId
     * @return Stops a job with a savepoint. Optionally, it can also emit a MAX_WATERMARK before taking the savepoint to flush out any state waiting for timers to fire. This async operation would return a 'triggerid' for further query identifier.
     */
    public static String stopFlinkJobWithSavepoint(String host, String port, String jobId, Object params) {
        return HttpUtils.httpPost(baseUrl(host, port, "jobs", jobId, "stop"), params);
    }

    /**
     * @param host  主机地址
     * @param port  端口地址
     * @param jarId Flink的jar包Id
     * @return 返回flink jobid
     */
    public static String startUpFlinkJob(String host, String port, String jarId, Properties properties) {
        String params = "";
        if (properties != null) {
            params = properties.entrySet().stream().map(x ->
                    x.getKey().toString() + "=" + x.getValue().toString()
            ).reduce((x, y) -> x + "&" + y).orElse("");
        }
        return HttpUtils.httpPost(baseUrl(host, port, "jars", jarId, "run") + "?" + params);
    }

    /**
     * 启动一个flink job
     * {"jobid":"05b1a1b22f7139f42fa6c60370b47a4e"}
     *
     * @param host   host
     * @param port   port
     * @param jarId  jarId
     * @param params 启动参数
     * @return {"jobid":"05b1a1b22f7139f42fa6c60370b47a4e"}
     */
    public static String startUpFlinkJob(String host, String port, String jarId, Object params) {
        return HttpUtils.httpPost(baseUrl(host, port, "jars", jarId, "run"), params);
    }

    /**
     * 获取flink jobManager的job list列表
     *
     * @param host 主机名
     * @param port 端口号
     * @return 返回flink job list jobs overview over all jobs and their current state.
     */
    public static String getListOfFlinkJob(String host, String port) {
        return HttpUtils.httpGetBody(baseUrl(host, port, "jobs")).toString();
    }

    /**
     * 获取flink job的详细信息
     *
     * @param host 主机地址名
     * @param port 端口号
     * @return 返回flink 任务的详细json信息
     */
    public static String getFlinkJobOverview(String host, String port) {
        return HttpUtils.httpGetBody(baseUrl(host, port, "jobs/overview")).toString();
    }

    /**
     * 获取flink jobmap 信息
     *
     * @param host 主机名
     * @param port 端口号
     * @return 获取flink jobs的hashMap 其中 hashMap的key是flink任务的id value是flink任务的运行状态
     */
    public static HashMap<String, String> getFlinkJobMap(String host, String port) {
        String jobListJson = HttpUtils.httpGetBody(baseUrl(host, port, "jobs")).toString();
        List<HashMap<String, String>> listHashMap = JsonPath.read(jobListJson, "$.jobs[*]");
        HashMap<String, String> resultHashMap = new HashMap<>();
        listHashMap.forEach(x -> resultHashMap.put(x.get("id"), x.get("status")));
        return resultHashMap;
    }

    /**
     * 获取flink 所有的jar包的信息
     *
     * @param host 主机名
     * @param port 端口号
     * @return 返回flink的jar包的信息
     */
    public static String getFlinkJarsInfo(String host, String port) {
        return HttpUtils.httpGetBody(baseUrl(host, port, "jars")).toString();
    }

    /**
     * 查询flink的savepoints操作状态
     * /jobs/:jobid/savepoints/:triggerid
     *
     * @param host      主机
     * @param port      端口
     * @param jobId     jobId
     * @param triggerId triggerId
     * @return Returns the status of a savepoint operation.
     */
    public static String getStatusOfSaOp(String host, String port, String jobId, String triggerId) {
        return HttpUtils.httpGetBody(baseUrl(host, port, "jobs", jobId, "savepoints", triggerId)).toString();
    }

    /**
     * 触发savepoint并可以选择是否cancel该job <br/>
     * 触发会返回一个triggerid用来查询状态, IN_PROGRESS -> COMPLETED <br/>
     * 返回示例: {"request-id":"cc798b2bb0148286d1a45e31b5f698a4"}
     *
     * @param host   host
     * @param port   port
     * @param jobId  jobid
     * @param params 参数
     * @return Triggers a savepoint, and optionally cancels the job afterwards. This async operation would return a 'triggerid' for further query identifier.
     */
    public static String triggerSavepoint(String host, String port, String jobId, Object params) {
        String url = baseUrl(host, port, "jobs", jobId, "savepoints");
        System.out.println("POST URL IS: -> " + url);
        return HttpUtils.httpPost(url, params);
    }

}
